package com.dayu.finecomm.device;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

import javax.annotation.Resource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.dayu.finecomm.decoder.HdTcpDecoder;
import com.dayu.finecomm.handler.tpacket.TPacketCollectHandler;
import com.dayu.finecomm.handler.tpacket.TPacketCommitHandler;
import com.dayu.finecomm.handler.tpacket.TPacketPushHandler;
import com.dayu.finecomm.handler.tpacket.TPacketStoreHandler;
import com.dayu.finecomm.mongo.service.MongoDAO;
import com.dayu.finecomm.pojo.DeviceConfig;
import com.dayu.finecomm.pojo.DevicePusher;
import com.dayu.finecomm.protocol.Protocols;
import com.dayu.finecomm.protocol.SpecialCareChannelInboundAdapter;
import com.dayu.finecomm.protocol.codec.Drive;
import com.dayu.finecomm.protocol.codec.ProtocolDecoder;
import com.dayu.finecomm.protocol.log.ProtocolLogger;
import com.dayu.finecomm.protocol.log.UnProtocolLogger;
import com.dayu.finecomm.protocol.pojo.TPacket;
import com.dayu.finecomm.source.CollectSource;
import com.dayu.finecomm.utils.Helpers;

import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.ReadTimeoutException;
import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;

/**
 * 纺织机
 * 
 * @author daiweiguo
 * 
 */
@Drive(CanDevice.DRIVE)
public class CanDevice extends ChannelInitializer<NioSocketChannel> implements Device {

	@Resource
	MongoDAO dao;

	static final String DRIVE = "can";

	static byte[] command = { 0x3c, 0x00, 0x00, 0x02, 0x00, 0x00, 0x00, 0x3e };

	static Logger LOGGER = LoggerFactory.getLogger(CanDevice.class);

	private Map<String, ChannelHandlerContext> listeners;

	final Set<String> protocols;
	private Set<InetSocketAddress> distributors;
	private Collection<DevicePusher> pushers;

	final DeviceConfig config;
	final String prefix;
	private CollectSource stations;
	private String listenType;

	private List<ProtocolDecoder<TPacket>> protocolDecoders;
	Map<String, byte[]> IDs;

	HashedWheelTimer timer = new HashedWheelTimer();

	public CanDevice(CollectSource stations, Set<String> protocols, DeviceConfig config) {
		this.protocols = protocols;
		this.prefix = config.prefix;
		this.stations = stations;
		this.config = config;

		if (this.config.idle == 0) {
			this.config.idle = 600;
		}
		protocolDecoders = Protocols.protocols(this.protocols);
		IDs = new ConcurrentHashMap<>();
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {

		IDs.clear();
		super.channelInactive(ctx);
	}

	private String getSessionId(ChannelHandlerContext ctx) {
		return Helpers.host(ctx);
	}

	@Override
	protected void initChannel(NioSocketChannel ctx) throws Exception {
		// 获取过程
		ChannelPipeline pipeline = ctx.pipeline();
		// 报文解析
		pipeline.addLast(new HdTcpDecoder(this.prefix, protocolDecoders));
		// 报文分发（作为主服务器时，如果收到报文，则分发给订阅了数据的网关，发送格式3种格式）
		pipeline.addLast(new TPacketCommitHandler(this));
		// 存储报文到站点集合
		pipeline.addLast(new TPacketCollectHandler(stations, false, config.name));
		// 报文存储
		pipeline.addLast(new TPacketStoreHandler());
		// 报文推送
		pipeline.addLast(new TPacketPushHandler(this, config));
		pipeline.addLast(new ProtocolLogger());
		pipeline.addLast(new UnProtocolLogger());

		if (this.config.idle > 0) {

			// Max idle time 600s
			pipeline.addFirst(new ReadTimeoutHandler(this.config.idle));
		}
		// Finalize
		pipeline.addLast(new SpecialCareChannelInboundAdapter<TPacket>() {

			@Override
			public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {

				if (cause instanceof ReadTimeoutException) {
					IDs.remove(getSessionId(ctx));
				} else if (cause instanceof IOException) {
					LOGGER.info("Force close by remote: " + " " + new String(IDs.get(getSessionId(ctx))));
					IDs.remove(getSessionId(ctx));
				} else {
					super.exceptionCaught(ctx, cause);
				}
			}

			@Override
			protected void channelRead0(ChannelHandlerContext ctx, TPacket msg) throws Exception {
				// TODO Auto-generated method stub
				// 延时执行查询
				LOGGER.info("channelRead Sendcommand Thead");
				timer.newTimeout(new TimerTask() {
					public void run(Timeout timeout) throws Exception {
						LOGGER.info("Send command");
						ctx.writeAndFlush(Unpooled.copiedBuffer(command));
					}
				}, config.interval, TimeUnit.SECONDS);
			}
		});

		ctx.writeAndFlush(Unpooled.copiedBuffer(command));

		LOGGER.info("connect client:" + ctx.remoteAddress());
	}

	@Override
	public Collection<DevicePusher> getPushers() {
		return pushers;
	}

	@Override
	public void setPushers(Collection<DevicePusher> pushers) {
		this.pushers = pushers;
	}

	@Override
	public Map<String, ChannelHandlerContext> getListeners() {
		return listeners;
	}

	@Override
	public void setListeners(Map<String, ChannelHandlerContext> lITENERS) {
		listeners = lITENERS;
	}

	@Override
	public Set<InetSocketAddress> getDistributors() {
		return distributors;
	}

	@Override
	public void setDistributors(Set<InetSocketAddress> distributors) {
		this.distributors = distributors;
	}

	@Override
	public String getListenType() {
		return listenType;
	}

	@Override
	public void setListenType(String listenType) {
		this.listenType = listenType;
	}
}